package com.cotte.flow.kafka;

import java.util.HashMap;
import java.util.Map;

import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import com.cotte.flow.service.BrushCardService;

import net.sf.json.JSONObject;

/**
 * 获取mes刷卡表数据
* <p>Title: ConsumerMes</p>  
* <p>Description: </p>  
* @author YGJ  
* @date 2018年12月4日
 */
@Component
public class ConsumerMes {
	
	protected final Logger logger = LoggerFactory.getLogger(this.getClass());
	
	@Autowired
	private BrushCardService service;
	
	@KafkaListener(group = "${mes.kafka.consumer.group.id}",topics = {"${mes.kafka.consumer.topic}"})
	public void consumerMes(ConsumerRecord<String, GenericRecord> record) {
		Object recordAfter = record.value().get("after");
		Map<String,String> map = new HashMap<String,String>();
		if(recordAfter != null){
			JSONObject jsonObject = JSONObject.fromObject(recordAfter.toString());
			String odid = jsonObject.get("Odid").toString();
			String stepCode = jsonObject.get("StepCode").toString();
			String empCode = jsonObject.get("EmpCode").toString();
			String syscode = jsonObject.get("SysCode").toString();
			String orderCode = jsonObject.get("OrderCode").toString();
			String loadDate = jsonObject.get("LoadDate").toString();
			map.put("odid", odid);
			map.put("stepCode", stepCode);
			map.put("empCode", empCode);
			map.put("syscode", syscode);
			map.put("orderCode", orderCode);
			map.put("loadDate", loadDate);
			service.saveBrushCard(map);
		}
	}
}
